package com.etl.component.file;

import cn.hutool.poi.excel.ExcelReader;
import cn.hutool.poi.excel.ExcelUtil;
import com.google.common.collect.Lists;
import com.component.api.constants.ComponentCodeConstants;
import com.component.api.constants.ComponentConstants;
import com.component.api.excepiton.ComponentException;
import com.component.api.model.PortData;
import com.component.api.model.ProcessResult;
import com.component.api.model.data.ColumnData;
import com.component.api.model.data.SetData;
import com.component.api.model.dto.FileCenterConfigDTO;
import com.component.api.model.param.ParamPair;
import com.component.api.model.param.TableColumn;
import com.etl.component.common.AbstractFlinkComponent;
import com.etl.component.file.center.FileCenter;
import com.etl.component.file.center.FileCenterManager;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;
import org.jetbrains.annotations.NotNull;

import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 描述：
 *  文件读取都是读出来都是字符串类型的
 * @author xianggj
 * @Date 2021/10/12 18:10
 **/
public class ExcelReadComponent extends AbstractFlinkComponent {

    @Override
    public ProcessResult process(List<PortData> datas, List<ParamPair> paramPairs) {
        Map<String, Object> params = paramToObject(paramPairs);
        //文件中心路径
        String path = (String)params.get(ComponentConstants.PATH);
        String type = (String)params.get(ComponentConstants.TYPE);
        InputStream inputStream = null;
        try {
            FileCenterConfigDTO config = (FileCenterConfigDTO)params.get(ComponentConstants.CONFIG);
            inputStream = FileCenterManager.getFileCenter(FileCenter.MINIO)
                    .getFileStream(config, path);
        } catch (Exception e) {
            e.printStackTrace();
            throw new ComponentException("文件读取失败", e);
        }
        DataSet<Row>  input = null;
        List<String> headers = (List<String>) params.get(ComponentConstants.HEADER);
        List<TableColumn> columns = getTableColumns(headers);
        if (ComponentConstants.XLS.equalsIgnoreCase(type) || ComponentConstants.XLSX.equalsIgnoreCase(type)){
            Boolean isFirstHeader = (Boolean) params.get(ComponentConstants.IS_FIRST_LINE);
            boolean isHeader = isFirstHeader == null? false: isFirstHeader;
            ExcelReader reader = ExcelUtil.getReader(inputStream);
            List<List<Object>> read = null;
            if (isHeader) {
                read = reader.read(1);
            } else {
                read = reader.read();
            }
            if (read == null || read.size() == 0){
                throw new ComponentException("不支持空文件读取");
            }

            final ExecutionEnvironment env = getEnv();
            TypeInformation[] fieldTypeInfos = new TypeInformation[headers.size()];
            Arrays.fill(fieldTypeInfos, BasicTypeInfo.STRING_TYPE_INFO);
            List<Row> rowList = new ArrayList<>();
            for (List<Object> list : read) {
                list = list.stream().map(e->e==null?null:e.toString()).collect(Collectors.toList());
                Row of = Row.of(list.toArray());
                rowList.add(of);
            }
            RowTypeInfo rowTypeInfo = new RowTypeInfo(fieldTypeInfos, headers.toArray(new String[0]));
            input = env.fromCollection(rowList, rowTypeInfo);
        } else {
            throw new ComponentException("文件格式错误");

        }
        return execute(input, columns, "执行excel拉取操作");
    }

    @Override
    public List<PortData> previewOutputColumns(List<PortData> datas, List<ParamPair> paramPairs) {
        Map<String, Object> params = paramToObject(paramPairs);
        List<String> headers = (List<String>) params.get(ComponentConstants.HEADER);
        List<TableColumn> columns = getTableColumns(headers);
        return Lists.newArrayList(new PortData(getRule(), new ColumnData(columns)));
    }

    @NotNull
    private List<TableColumn> getTableColumns(List<String> headers) {
        List<TableColumn> columns = new ArrayList<>();
        for (int i = 0; i < headers.size(); i++) {
            String header = headers.get(i);
            columns.add(new TableColumn(header, ComponentConstants.VARCHAR, ComponentConstants.DEFAULT_VARCHAR_LENGTH));
        }
        return columns;
    }

    @Override
    public String getRule() {
        return ComponentCodeConstants.XLS_PARSING;
    }
}
